package com.cxs.listener;

import org.springframework.stereotype.Component;

/**
 * @Author: cxs
 */
@Component
public class ASimpleListener {

    ////////////// 自动提交偏移量 ///////////////
    /**
     * 一个主题可以有多个分区，每个分区对应一个消费者
     * 例如有3个分区，那么一个消费者组内最多可以有三个消费者进行消费
     * 也就是 分区数量 >= 一个组内消费者的数量，多的消费者没有消息消费
     * 测试时kafka有一定的延迟
     *
     * 报错也提交了
     *
     * @param record
     */
//    @KafkaListener(id = "cxs-id1", groupId = "cxs-g2", topics = {"my-topic7"})
//    public void simpleMsgListener(ConsumerRecord<?, ?> record) {
//        System.out.println("cxs-id1");
//        System.out.println(record.topic());
//        System.out.println(record.partition());
//        System.out.println(record.headers());
//        System.out.println(record.value());
//        System.out.println(record.toString());
//    }
//
//
//
//    @KafkaListener(id = "cxs-id2", groupId = "cxs-g2", topics = {"my-topic7"})
//    public void simpleMsgListener2(ConsumerRecord<?, ?> record) {
//        System.out.println("cxs-id2");
//        System.out.println(record.topic());
//        System.out.println(record.partition());
//        System.out.println(record.headers());
//        System.out.println(record.value());
//        System.out.println(record.toString());
//    }


//    @KafkaListener(id = "cxs-id3", groupId = "cxs-g2", topics = {"my-topic7"})
//    public void simpleMsgListener3(ConsumerRecord<?, ?> record) {
//        System.out.println("cxs-id3");
//        System.out.println(record.topic());
//        System.out.println(record.partition());
//        System.out.println(record.headers());
//        System.out.println(record.value());
//        System.out.println(record.toString());
//    }


    ///////////////// batch 模式 /////////////////

    /**
     * spring.kafka.listener.type: batch 批量拉取消息消费
     * spring.kafka.consumer.max-poll-records: 10 批量消费每次最多消费多少条消息  不指定默认一次拉取500条
     *
     * @param record
     */
//    @KafkaListener(id = "cxs-id1", groupId = "cxs-g2", topics = {"my-topic7"})
//    public void simpleMsgListener(List<ConsumerRecord<?, ?>> record) {
//        System.out.println("cxs-id1");
//        System.out.println(record.toString());
//    }
//
//
//    @KafkaListener(id = "cxs-id2", groupId = "cxs-g2", topics = {"my-topic7"})
//    public void simpleMsgListener2(List<ConsumerRecord<?, ?>> record) {
//        System.out.println("cxs-id2");
//        System.out.println(record.toString());
//    }


    ///////////// 手动签收 ///////////////

    /**
     * spring.kafka.consumer.enable-auto-commit: false
     * spring.kafka.listener.ack-mode: manual #在处理完最后一次轮询的所有结果后，将队列排队，并在一次操作中提交偏移量。可以认为是在批处理结束时提交偏移量
     * spring.kafka.listener.ack-mode: manual_immediate ##只要在侦听器线程上执行确认，就立即提交偏移。会在批量执行的时候逐一提交它们
     * kafka不支持重试消费，一般可以 使用try catch方式 写业务逻辑 出现异常可以不签收消息
     * 也可以使用kafka自带的消费异常处理机制 ConsumerAwareListenerErrorHandler
     */
//    @KafkaListener(id = "cxs-id1", groupId = "cxs-g2", topics = {"my-topic7"}, errorHandler = "consumerAwareErrorHandler")
//    public void simpleMsgListener(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
//        System.out.println("线程id:" + Thread.currentThread().getId());
//        System.out.println("cxs-id1");
//        System.out.println(ack);
//        System.out.println(records.toString());
//        // 如果是批量消息， 则视为一个整体，一个报错，全部都会去errorHandler
//        // 在errorHandler里面还要是consumer.commitSync() 则这五条消息都被签收了
//        int i = 10 / 0;
//        ack.acknowledge();
//    }


//    @KafkaListener(id = "cxs-id2", groupId = "cxs-g2", topics = {"my-topic7"})
//    public void simpleMsgListener2(List<ConsumerRecord<?, ?>> record, Acknowledgment ack) {
//        System.out.println("线程id:" + Thread.currentThread().getId());
//        System.out.println("cxs-id2");
//        System.out.println(ack);
//        System.out.println(record.toString());
//        ack.acknowledge();
//    }


    ///////////// 并发消费 //////////////

    /**
     * spring.kafka.listener.concurrency: 3  # 为每个消费者实例创建多少个线程。多出分区的线程空闲
     * 这个并发消费理解起来很有歧义
     * 正确的理解是
     * 如果我只有一个组，并且只有一个消费者，那么就开启3个线程去消费消息，但是
     * 如果我只有2个分区，那么多的一个线程会空闲，也就是说，一个分区只对应一个线程
     * 但是我如果有一个组，组内有两个消费者，刚好对应两个分区，那么这个线程设置也就没有意义了
     *
     * @param record
     * @param ack
     */
//    @KafkaListener(id = "cxs-id1", groupId = "cxs-g2", topics = {"my-topic7"})
//    public void simpleMsgListener(List<ConsumerRecord<?, ?>> record, Acknowledgment ack) {
//        System.out.println("线程id:" + Thread.currentThread().getId());
//        System.out.println("cxs-id1");
//        System.out.println(ack);
//        System.out.println(record.toString());
//        ack.acknowledge();
//    }

    /////////////////// 消息过滤 ///////////////

    /**
     * 消息过滤器可以在消息抵达consumer之前被拦截，
     * 在实际应用中，我们可以根据自己的业务逻辑，筛选出需要的信息再交由KafkaListener处理
     * 不需要的消息则过滤掉
     *
     * @param records
     * @param ack
     */
//    @KafkaListener(groupId = "cxs-g2", topics = {"my-topic7"}, filter = "recordFilterStrategy")
//    public void simpleMsgListener(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
//        System.out.println("开始消费了");
//        System.out.println(records);
//        ack.acknowledge();
//    }


    /////////// 消息转发 ////////////////

    /**
     * 应用A从TopicA获取到消息，经过处理后转发到TopicB，再由应用B监听处理消息
     * 只需要通过一个@SendTo注解，被注解方法的return值即转发的消息内容
     * 一般用于单个消息  我非要试试批量消息转发
     *
     * @param records
     * @param ack
     */
//    @KafkaListener(groupId = "cxs-g2", topics = {"my-topic7"})
//    @SendTo(value = "my-topic8")
//    public String simpleMsgListener(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
//        System.out.println("开始消费了");
//        System.out.println(records);
//        ack.acknowledge();
//        return records.toString();
//    }

    // todo 定时启动、停止监听器  延迟消息


}
